package avicit.bdp.dcs.task.service;

import avicit.bdp.common.utils.BdpLogUtil;
import avicit.bdp.common.utils.CheckParaUtils;
import avicit.bdp.common.utils.database.ColumnInfo;
import avicit.bdp.common.utils.database.IotDbSessionUtils;
import avicit.bdp.common.utils.enums.JobType;
import avicit.bdp.core.constant.Constants;
import avicit.bdp.core.util.json.JSONUtils;
import avicit.bdp.dcs.datasource.dto.JobDatasource;
import avicit.bdp.dcs.datasource.service.BaseDbService;
import avicit.bdp.dcs.datasource.service.DataSourceUtils;
import avicit.bdp.dcs.datasource.service.DatasourceQueryService;
import avicit.bdp.dcs.datasource.service.TableService;
import avicit.bdp.dcs.detail.dto.TaskDataDetailDTO;
import avicit.bdp.dcs.detail.dto.TaskTableDetailDTO;
import avicit.bdp.dcs.detail.service.TaskDataDetailService;
import avicit.bdp.dcs.detail.service.TaskTableDetailService;
import avicit.bdp.dcs.job.dto.JobDTO;
import avicit.bdp.dcs.json.service.DataxJsonService;
import avicit.bdp.dcs.stream.dto.MsgTemplateDto;
import avicit.bdp.dcs.stream.service.MsgTemplateService;
import avicit.bdp.dcs.task.dto.ColumnMappingDto;
import avicit.bdp.dcs.task.dto.StreamColumnMappingDto;
import avicit.bdp.dcs.task.dto.TableMappingDto;
import avicit.bdp.dcs.task.dto.TaskDTO;
import avicit.bdp.dcs.tools.enums.Enable;
import avicit.bdp.dcs.tools.enums.ProcessMode;
import avicit.bdp.dcs.tools.enums.SyncMode;
import avicit.bdp.dcs.tools.enums.TaskDetailStatus;
import avicit.bdp.dcs.utils.DcsConstants;
import avicit.bdp.dcs.utils.DcsErrorCode;
import avicit.bdp.dcs.utils.JdbcConstants;
import avicit.platform6.commons.utils.ComUtil;
import avicit.platform6.core.exception.BusinessException;
import com.alibaba.fastjson2.JSONArray;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * @金航数码科技有限责任公司
 * @作者：liyb
 * @邮箱：liyb@avic-digital.com
 * @创建时间： 2020-09-08 15:23
 * @类说明： 流式任务服务类
 * @修改记录：
 */
@Service
@Transactional(rollbackFor = Exception.class)
public class StreamTaskService implements TaskInterface {

  private static final Logger logger = LoggerFactory.getLogger(StreamTaskService.class);

  @Autowired private TaskService taskService;
  @Autowired private DatasourceQueryService datasourceQueryService;
  @Autowired private DataxJsonService dataxJsonService;
  @Autowired private TableService tableService;
  @Autowired private TaskTableDetailService taskTableDetailService;
  @Autowired private TaskDataDetailService taskDataDetailService;
  @Autowired private MsgTemplateService msgTemplateService;
  @Autowired private DataSourceUtils dataSourceUtils;
  @Autowired private BaseDbService baseDbService;

  /** 查看任务 */
  @Override
  public TaskDTO getTaskById(JobDTO jobDTO, String srcTableName) throws Exception {
    // 1、查询任务
    TaskDTO query = new TaskDTO();
    query.setJobId(jobDTO.getId());
    query.setFlag(Enable.ZERO.getCode());
    TaskDTO task = taskService.selectOne(query);
    CheckParaUtils.checkDbObjectValidity(
        task, logger, String.format(DcsErrorCode.JOB_NULL.getDesc(), jobDTO.getId()));
    task.setDataxJson(null);

    // 流式采集
    List<TableMappingDto> tableMappingDtoList =
        JSONUtils.toList(jobDTO.getSelectedTable(), TableMappingDto.class);
    Map<String, List<ColumnInfo>> srcColumnMap = new HashMap<>(16);
    Map<String, List<ColumnInfo>> dstColumnMap = new HashMap<>(16);
    if (CollectionUtils.isNotEmpty(tableMappingDtoList)) {
      for (TableMappingDto tableMappingDto : tableMappingDtoList) {
        String templateId = tableMappingDto.getSrc_table();

        // 2、查询源表列字段
        List<ColumnInfo> srcColumnList =
            datasourceQueryService.getKafkaTemplateColumnList(templateId);

        // 根据源表字段根据列映射查询字段前缀信息
        setColumnPrefix(srcColumnList, templateId, task);
        srcColumnMap.put(templateId, srcColumnList);

        // 3、查询目标列字段
        List<ColumnInfo> dstColumnList = new ArrayList<>();
        JobDatasource dstDatasource =
            dataSourceUtils.convertDatasource(jobDTO.getDstDatasourceId());
        if (JdbcConstants.IOTDB.equals(dstDatasource.getDbType())) {
          IotDbSessionUtils.initIotdbSessionPool(
              dstDatasource.getJdbcUrl(),
              dstDatasource.getJdbcUsername(),
              dstDatasource.getEncryptJdbcPassword());
          dstColumnList = IotDbSessionUtils.getColumnListByDevice(tableMappingDto.getDst_table());
        } else if (JdbcConstants.HIVE.equals(dstDatasource.getDbType())) {
          dstColumnList =
              baseDbService.getColumnList(
                  jobDTO.getDstDatasourceId(), tableMappingDto.getDst_table());

          // 如果数据源为hive，需要将列名转小写完成自动列映射
          if (CollectionUtils.isEmpty(dstColumnList)) {
            logger.info("数据源为hive，需要将列名转小写完成自动列映射");
            dstColumnList = srcColumnList;
            dstColumnList =
                dstColumnList.stream()
                    .map(
                        columnInfo -> {
                          ColumnInfo dstColumnInfo = new ColumnInfo();
                          BeanUtils.copyProperties(columnInfo, dstColumnInfo);
                          dstColumnInfo.setName(dstColumnInfo.getName().toLowerCase());
                          return dstColumnInfo;
                        })
                    .collect(Collectors.toList());
          }
        }

        if (CollectionUtils.isNotEmpty(dstColumnList)) {
          // 目标列按源列顺序排列
          List<StreamColumnMappingDto> columnMappingDtoList =
              JSONUtils.toList(task.getMappingColumn(), StreamColumnMappingDto.class);
          if (CollectionUtils.isNotEmpty(columnMappingDtoList)) {
            for (StreamColumnMappingDto streamColumnMappingDto : columnMappingDtoList) {
              if (templateId.equals(streamColumnMappingDto.getTemplateId())) {
                List<ColumnMappingDto> columnList = streamColumnMappingDto.getColumnList();
                dstColumnList =
                    taskService.sortBySrcColumn(
                        srcColumnList, dstColumnList, JSONArray.toJSONString(columnList));
                break;
              }
            }
          }

          dstColumnMap.put(tableMappingDto.getDst_table(), dstColumnList);
        } else {
          dstColumnMap.put(tableMappingDto.getDst_table(), srcColumnList);
        }
      }
    }

    task.setSrcColumnMap(srcColumnMap);
    task.setDstColumnMap(dstColumnMap);
    return task;
  }

  /** 更新任务 */
  @Override
  public String updateTask(TaskDTO oldTask, TaskDTO taskDTO) throws Exception {
    if (StringUtils.isNotEmpty(taskDTO.getMappingColumn())) {
      List<StreamColumnMappingDto> streamColumnMappingDtoList =
          JSONUtils.toList(taskDTO.getMappingColumn(), StreamColumnMappingDto.class);

      // 判断列名称合法性
      this.checkStreamColumnName(streamColumnMappingDtoList);

      // 先更新列映射关系，再生成dataxjson
      taskService.updateByPrimaryKeySelective(taskDTO);
      return dataxJsonService.buildDataxJson(oldTask);
    }
    return null;
  }

  /** 批量保存任务 一个job关联一个task */
  @Override
  public void saveTasksByJobId(JobDTO jobDTO) throws Exception {
    // 根据jobId查询dcs_task,如果有记录则修改，没有则新增
    TaskDTO query = new TaskDTO();
    query.setJobId(jobDTO.getId());
    query.setFlag(Enable.ZERO.getCode());
    TaskDTO taskDTO = taskService.selectOne(query);
    TableMappingDto tableMappingDto = getStreamSrcDstTableName(jobDTO);
    if (null == taskDTO) {
      String mappingColumn = this.getMappingColumnString(jobDTO);

      // 1、新增task记录
      taskDTO = new TaskDTO();
      taskDTO.setId(ComUtil.getId());
      taskDTO.setJobId(jobDTO.getId());
      taskDTO.setJobType(jobDTO.getJobType());
      taskDTO.setSrcTableName(tableMappingDto.getSrc_table());
      taskDTO.setDstTableName(tableMappingDto.getDst_table());
      taskDTO.setMappingColumn(mappingColumn);
      taskDTO.setSyncMode(SyncMode.TOTAL_SYNC.getCode());
      taskDTO.setFlag(Enable.ZERO.getCode());

      taskService.insert(taskDTO);

      // 记录日志
      BdpLogUtil.log4Insert(taskDTO);
    } else {
      // 2、修改task记录,如果源表和目标表未变则不更新
      List<StreamColumnMappingDto> streamColumnMappingDtoList =
          JSONUtils.toList(taskDTO.getMappingColumn(), StreamColumnMappingDto.class);

      // 判断表同步信息是否变更
      String mappingColumn = getMappingColumnString(streamColumnMappingDtoList, jobDTO);
      taskDTO.setSrcTableName(tableMappingDto.getSrc_table());
      taskDTO.setDstTableName(tableMappingDto.getDst_table());
      taskDTO.setMappingColumn(mappingColumn);
      taskService.updateByPrimaryKeySelective(taskDTO);

      // 2.1删除上次任务保存bdp_dcs_task_table_detail表数据和保存bdp_dcs_task_data_detail表数据
      TaskTableDetailDTO taskTableDetailDTO =
          TaskTableDetailDTO.builder().taskId(taskDTO.getId()).build();
      taskTableDetailService.delete(taskTableDetailDTO);

      TaskDataDetailDTO taskDataDetailDTO =
          TaskDataDetailDTO.builder().taskId(taskDTO.getId()).build();
      taskDataDetailService.delete(taskDataDetailDTO);
    }

    // 3、保存bdp_dcs_task_table_detail表数据和保存bdp_dcs_task_data_detail表数据
    taskService.saveTaskDetailData(
        tableMappingDto.getSrc_table(), tableMappingDto.getDst_table(), taskDTO, jobDTO);
  }

  /**
   * 创建目标表并回写建表sql语句
   *
   * @param jobDTO jobDTO
   * @param taskDTO taskDTO
   * @return 创建目标表结果
   * @throws Exception
   */
  @Override
  public Boolean createTable(JobDTO jobDTO, TaskDTO taskDTO) throws Exception {
    boolean createTableFlag = true;
    JobDatasource dstDatasource = dataSourceUtils.convertDatasource(jobDTO.getDstDatasourceId());
    logger.info("processMode:{}", jobDTO.getProcessMode());

    List<StreamColumnMappingDto> streamColumnMappingDtoList =
        JSONUtils.toList(taskDTO.getMappingColumn(), StreamColumnMappingDto.class);
    if (CollectionUtils.isNotEmpty(streamColumnMappingDtoList)) {
      for (StreamColumnMappingDto streamColumnMappingDto : streamColumnMappingDtoList) {

        // 目标已存在表的处理模式，预检查并报错拦截：0，覆盖目标表：1
        if (jobDTO.getProcessMode() == ProcessMode.PRECHECK_ERROR.getCode()) {
          if (taskService.isExistTable(
              jobDTO.getDstDatasourceId(), streamColumnMappingDto.getTableName())) {
            logger.info(
                DcsErrorCode.DST_DB_EXIST_SAME_TABLE.getDesc(),
                dstDatasource.getJdbcUrl(),
                streamColumnMappingDto.getTableName());
            throw new BusinessException(
                String.format(
                    DcsErrorCode.DST_DB_EXIST_SAME_TABLE.getDesc(),
                    dstDatasource.getJdbcUrl(),
                    streamColumnMappingDto.getTableName()));
          }
        } else if (jobDTO.getProcessMode() == ProcessMode.COVER_DST_TABLE.getCode()) {
          boolean isDel =
              baseDbService.dropTable(
                  jobDTO.getDstDatasourceId(), streamColumnMappingDto.getTableName());
          logger.info("删除目标表:{},删除结果:{}", streamColumnMappingDto.getTableName(), isDel);

          taskService.saveTableLog(
              taskDTO,
              String.format("删除目标表:%s,删除结果:%s", streamColumnMappingDto.getTableName(), isDel));
        } else if (jobDTO.getProcessMode() == ProcessMode.RETAIN_DST_TABLE.getCode()) {
          logger.info("已存在表的处理模式为保留目标表，不做处理");
          return false;
        } else {
          throw new BusinessException("不支持的已存在表的处理模式");
        }
      }
    }

    StringBuilder buf = new StringBuilder();
    if (CollectionUtils.isNotEmpty(streamColumnMappingDtoList)) {
      for (StreamColumnMappingDto streamColumnMappingDto : streamColumnMappingDtoList) {

        // 1、列名称转换
        Map<String, String> columnMap = new HashMap<>(16);
        List<ColumnMappingDto> columnList = streamColumnMappingDto.getColumnList();
        if (CollectionUtils.isNotEmpty(columnList)) {
          for (ColumnMappingDto columnMappingDto : columnList) {
            columnMap.put(columnMappingDto.getSrcColumnName(), columnMappingDto.getDstColumnName());
          }
        }

        // 2、循环创建表
        Map<String, String> map =
            baseDbService.createTable(
                jobDTO,
                streamColumnMappingDto.getTemplateId(),
                streamColumnMappingDto.getTableName(),
                columnMap);

        buf.append(map.get(DcsConstants.CREATE_TABLE_SQL)).append(";");

        taskService.saveTableLog(
            taskDTO, String.format("创建表sql:%s", map.get(Constants.CREATE_TABLE_SQL)));

        if (!Boolean.valueOf(map.get(Constants.SUCCESS))) {
          // 有一个表创建失败，总的结果为失败
          createTableFlag = false;

          taskService.saveTableLog(
              taskDTO, String.format("创建表失败,原因是:%s", map.get(Constants.MESSAGE)));
        } else {
          taskService.saveTableLog(taskDTO, String.format("创建表成功"));
        }
      }
    }

    // 3、回写创建表sql语句
    TaskTableDetailDTO query =
        TaskTableDetailDTO.builder()
            .taskId(taskDTO.getId())
            .srcTableName(taskDTO.getSrcTableName())
            .build();
    TaskTableDetailDTO tableDetailDTO = taskTableDetailService.selectOne(query);
    CheckParaUtils.checkDbObjectValidity(
        tableDetailDTO, logger, String.format("任务id:%s,没找到对应的表结构迁移明细数据", taskDTO.getId()));

    tableDetailDTO.setCreateSql(buf.toString());
    if (createTableFlag) {
      tableDetailDTO.setTaskStatus(TaskDetailStatus.COMPLETED.getCode());
      taskDTO.setIsCreated(Enable.ONE.getCode());
      taskService.updateByPrimaryKeySelective(taskDTO);
    } else {
      tableDetailDTO.setTaskStatus(TaskDetailStatus.EXECUTE_FAIL.getCode());
    }

    taskTableDetailService.updateByPrimaryKeySelective(tableDetailDTO);
    logger.info("创建表sql={}", buf.toString());
    logger.info("************创建表返回结果：{},状态为:{}", createTableFlag, tableDetailDTO.getTaskStatus());
    return createTableFlag;
  }

  private TableMappingDto getStreamSrcDstTableName(JobDTO jobDTO) {
    String selectedTable = jobDTO.getSelectedTable();
    TableMappingDto tableMappingDtoResult = new TableMappingDto();
    StringBuilder srcTableName = new StringBuilder();
    StringBuilder dstTableName = new StringBuilder();
    List<TableMappingDto> tableMappingList = JSONUtils.toList(selectedTable, TableMappingDto.class);
    for (TableMappingDto tableMappingDto : tableMappingList) {
      String templateId = tableMappingDto.getSrc_table();
      MsgTemplateDto msgTemplateDto = msgTemplateService.selectByPrimaryKey(templateId);
      srcTableName.append(msgTemplateDto.getDeviceName());
      srcTableName.append(Constants.COMMA);
      dstTableName.append(tableMappingDto.getDst_table());
      dstTableName.append(Constants.COMMA);
    }
    tableMappingDtoResult.setSrc_table(srcTableName.substring(0, srcTableName.length() - 1));
    tableMappingDtoResult.setDst_table(dstTableName.substring(0, dstTableName.length() - 1));
    return tableMappingDtoResult;
  }

  /** 获取列映射关系 */
  private String getMappingColumnString(JobDTO jobDTO) throws Exception {
    // 获取模板和表的对应关系
    List<TableMappingDto> tableMappingDtoList =
        JSONUtils.toList(jobDTO.getSelectedTable(), TableMappingDto.class);

    JobDatasource dstDatasource = dataSourceUtils.convertDatasource(jobDTO.getDstDatasourceId());

    // 遍历获取每个模板和表的对应关系
    List<StreamColumnMappingDto> streamColumnMappingDtoList = new ArrayList<>();
    if (CollectionUtils.isNotEmpty(tableMappingDtoList)) {
      for (TableMappingDto tableMappingDto : tableMappingDtoList) {
        Boolean isExistDstTable =
            taskService.isExistInDstDatabase(
                jobDTO.getDstDatasourceId(), tableMappingDto.getDst_table());
        ArrayList<ColumnMappingDto> columnMappingDtoList;

        if (isExistDstTable) {
          // 目标数据库存在目标表
          // 获取源表列名称以及类型
          List<ColumnInfo> srcColumnList =
              datasourceQueryService.getKafkaTemplateColumnList(tableMappingDto.getSrc_table());

          // 获取目标表列名称以及类型
          List<ColumnInfo> dstColumnList = new ArrayList<>();
          if (JdbcConstants.IOTDB.equals(dstDatasource.getDbType())) {
            IotDbSessionUtils.initIotdbSessionPool(
                dstDatasource.getJdbcUrl(),
                dstDatasource.getJdbcUsername(),
                dstDatasource.getEncryptJdbcPassword());
            dstColumnList = IotDbSessionUtils.getColumnListByDevice(tableMappingDto.getDst_table());
          } else if (JdbcConstants.HIVE.equals(dstDatasource.getDbType())) {
            dstColumnList =
                baseDbService.getColumnList(dstDatasource.getId(), tableMappingDto.getDst_table());
          } else {
            logger.info(DcsErrorCode.UNSUPPORTED_DB_TYPE.getDesc(), dstDatasource.getDbType());
          }
          columnMappingDtoList = taskService.getMergeColumnList(srcColumnList, dstColumnList);
        } else {
          // 目标数据库不存在目标表
          // 获取源表列名称以及类型
          List<ColumnInfo> mergeColumnList =
              datasourceQueryService.getKafkaTemplateColumnList(tableMappingDto.getSrc_table());
          columnMappingDtoList =
              taskService.getColumnMappingDtos(mergeColumnList, dstDatasource.getDbType());
        }

        StreamColumnMappingDto streamColumnMappingDto = new StreamColumnMappingDto();
        streamColumnMappingDto.setTemplateId(tableMappingDto.getSrc_table());

        // 获取模板名称
        MsgTemplateDto msgTemplateDto =
            msgTemplateService.selectByPrimaryKey(tableMappingDto.getSrc_table());
        streamColumnMappingDto.setTemplateName(msgTemplateDto.getDeviceName());
        streamColumnMappingDto.setTableName(tableMappingDto.getDst_table());
        streamColumnMappingDto.setColumnList(columnMappingDtoList);
        streamColumnMappingDtoList.add(streamColumnMappingDto);
      }
    }

    logger.info("列映射关系为:{}", JSONArray.toJSONString(streamColumnMappingDtoList));
    return JSONArray.toJSONString(streamColumnMappingDtoList);
  }

  /** 获取列映射关系 */
  private String getMappingColumnString(
      List<StreamColumnMappingDto> oldStreamColumnMappingDtoList, JobDTO jobDTO) throws Exception {
    List<TableMappingDto> tableMappingDtoList =
        JSONUtils.toList(jobDTO.getSelectedTable(), TableMappingDto.class);

    JobDatasource dstDatasource = dataSourceUtils.convertDatasource(jobDTO.getDstDatasourceId());

    // 遍历获取每个模板和表的对应关系
    List<StreamColumnMappingDto> newStreamColumnMappingDtoList = new ArrayList<>();
    if (CollectionUtils.isNotEmpty(tableMappingDtoList)) {
      for (TableMappingDto tableMappingDto : tableMappingDtoList) {

        List<StreamColumnMappingDto> streamColumnMappingList =
            oldStreamColumnMappingDtoList.stream()
                .filter(
                    oldStreamColumnMappingDto ->
                        tableMappingDto
                                .getSrc_table()
                                .equals(oldStreamColumnMappingDto.getTemplateId())
                            && tableMappingDto
                                .getDst_table()
                                .equals(oldStreamColumnMappingDto.getTableName()))
                .collect(Collectors.toList());

        if (CollectionUtils.isNotEmpty(streamColumnMappingList)) {
          newStreamColumnMappingDtoList.addAll(streamColumnMappingList);
        } else {
          Boolean isExist =
              taskService.isExistInDstDatabase(
                  jobDTO.getDstDatasourceId(), tableMappingDto.getDst_table());
          ArrayList<ColumnMappingDto> columnMappingDtoList;

          if (isExist) {
            // 目标数据库存在目标表
            // 获取源表列名称以及类型
            List<ColumnInfo> srcColumnList =
                datasourceQueryService.getKafkaTemplateColumnList(tableMappingDto.getSrc_table());

            // 获取目标表列名称以及类型
            List<ColumnInfo> dstColumnList = new ArrayList<>();
            if (JdbcConstants.IOTDB.equals(dstDatasource.getDbType())) {
              IotDbSessionUtils.initIotdbSessionPool(
                  dstDatasource.getJdbcUrl(),
                  dstDatasource.getJdbcUsername(),
                  dstDatasource.getEncryptJdbcPassword());
              dstColumnList =
                  IotDbSessionUtils.getColumnListByDevice(tableMappingDto.getDst_table());
            } else if (JdbcConstants.HIVE.equals(dstDatasource.getDbType())) {
              dstColumnList =
                  baseDbService.getColumnList(
                      dstDatasource.getId(), tableMappingDto.getDst_table());
            } else {
              // 不支持
              logger.debug("skip");
            }
            columnMappingDtoList = taskService.getMergeColumnList(srcColumnList, dstColumnList);
          } else {
            // 目标数据库不存在目标表
            // 获取源表列名称以及类型
            List<ColumnInfo> mergeColumnList =
                datasourceQueryService.getKafkaTemplateColumnList(tableMappingDto.getSrc_table());

            columnMappingDtoList =
                taskService.getColumnMappingDtos(mergeColumnList, dstDatasource.getDbType());
          }

          StreamColumnMappingDto streamColumnMappingDto = new StreamColumnMappingDto();
          streamColumnMappingDto.setTemplateId(tableMappingDto.getSrc_table());

          // 获取模板名称
          MsgTemplateDto msgTemplateDto =
              msgTemplateService.selectByPrimaryKey(tableMappingDto.getSrc_table());
          streamColumnMappingDto.setTemplateName(msgTemplateDto.getDeviceName());
          streamColumnMappingDto.setTableName(tableMappingDto.getDst_table());
          streamColumnMappingDto.setColumnList(columnMappingDtoList);
          newStreamColumnMappingDtoList.add(streamColumnMappingDto);
        }
      }
    }
    return JSONArray.toJSONString(newStreamColumnMappingDtoList);
  }

  /** 检查列名合法性 */
  private void checkStreamColumnName(List<StreamColumnMappingDto> streamColumnMappingDtoList) {
    if (CollectionUtils.isNotEmpty(streamColumnMappingDtoList)) {
      for (StreamColumnMappingDto streamColumnMappingDto : streamColumnMappingDtoList) {
        List<ColumnMappingDto> columnList = streamColumnMappingDto.getColumnList();
        if (CollectionUtils.isNotEmpty(columnList)) {
          for (ColumnMappingDto columnMappingDto : columnList) {
            if (!tableService.isValidColumnName(columnMappingDto.getDstColumnName())) {
              throw new BusinessException(
                  String.format(
                      DcsErrorCode.COLUMN_NAME_CHECK.getDesc(),
                      columnMappingDto.getDstColumnName()));
            }
          }
        }
      }
    }
  }

  @Override
  public Integer getType() {
    return JobType.STREAM.getCode();
  }

  /** 设置列前缀 */
  private void setColumnPrefix(List<ColumnInfo> srcColumnList, String templateId, TaskDTO task) {
    for (ColumnInfo columnInfo : srcColumnList) {
      List<ColumnMappingDto> columnMappingDtoList = getColumnInfoByTemplateId(task, templateId);
      for (ColumnMappingDto columnMappingDto : columnMappingDtoList) {
        if (columnInfo.getName().equals(columnMappingDto.getSrcColumnName())) {
          columnInfo.setPrefix(columnMappingDto.getSrcColumnPrefix());
        }
      }
    }
  }

  private List<ColumnMappingDto> getColumnInfoByTemplateId(TaskDTO task, String templateId) {
    List<ColumnMappingDto> list = new ArrayList<>();
    List<StreamColumnMappingDto> streamColumnMappingDtoList =
        JSONUtils.toList(task.getMappingColumn(), StreamColumnMappingDto.class);
    for (StreamColumnMappingDto streamColumnMappingDto : streamColumnMappingDtoList) {
      if (templateId.equals(streamColumnMappingDto.getTemplateId())) {
        list = streamColumnMappingDto.getColumnList();
        break;
      }
    }
    return list;
  }
}
